// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package org.apache.impala.calcite.service;

import org.apache.impala.analysis.StmtMetadataLoader.StmtTableCache;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.ParseException;
import org.apache.impala.common.UnsupportedFeatureException;

import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * UnsupportedChecker containts methods which determine if a feature
 * is unsupported for the Calcite planner.
 */
public class UnsupportedChecker {

  private static Pattern LEFT_SEMI = Pattern.compile(".*\\bleft\\ssemi\\b.*",
      Pattern.CASE_INSENSITIVE);

  private static Pattern RIGHT_SEMI = Pattern.compile(".*\\bright\\ssemi\\b.*",
      Pattern.CASE_INSENSITIVE);

  private static Pattern LEFT_ANTI = Pattern.compile(".*\\bleft\\santi\\b.*",
      Pattern.CASE_INSENSITIVE);

  private static Pattern RIGHT_ANTI = Pattern.compile(".*\\bright\\santi\\b.*",
      Pattern.CASE_INSENSITIVE);

  private static Pattern INPUT_FILE_NAME = Pattern.compile(".*\\binput__file__name\\b.*",
      Pattern.CASE_INSENSITIVE);

  private static Pattern FILE_POSITION = Pattern.compile(".*\\bfile__position\\b.*",
      Pattern.CASE_INSENSITIVE);

  private static Pattern TABLE_NOT_FOUND =
      Pattern.compile(".*\\bTable '(.*)' not found\\b.*", Pattern.CASE_INSENSITIVE);

  private static Pattern COLUMN_NOT_FOUND =
      Pattern.compile(".*\\bColumn '(.*)' not found\\b.*", Pattern.CASE_INSENSITIVE);

  public static void throwUnsupportedIfKnownException(Exception e)
      throws ImpalaException {
    String s = e.toString().replace("\n"," ");
    if (LEFT_ANTI.matcher(s).matches() || RIGHT_ANTI.matcher(s).matches()) {
      throw new UnsupportedFeatureException("Anti joins not supported.");
    }
    if (LEFT_SEMI.matcher(s).matches() || RIGHT_SEMI.matcher(s).matches()) {
      throw new UnsupportedFeatureException("Semi joins not supported.");
    }
    if (INPUT_FILE_NAME.matcher(s).matches() || FILE_POSITION.matcher(s).matches()) {
      throw new UnsupportedFeatureException("Virtual columns not supported.");
    }
  }

  public static void throwUnsupportedIfKnownException(Exception e,
      StmtTableCache stmtTableCache) throws ImpalaException {
    throwUnsupportedIfKnownException(e);
    String s = e.toString().replace("\n"," ");
    Matcher m = TABLE_NOT_FOUND.matcher(s);

    // If the error given is "table/column not found", it is possible that the message
    // was generated by a complex column that looks like a table
    // (e.g. mytbl.my_complex_column) which is currently not supported. We check for
    // this possibility by seeing if the 'table not found' is identified as a column
    // within one of the tables in the query. This check isn't fool-proof in that
    // it might actually be a table that also is a column name in another table.
    // However, that case should be extremely rare, and the result would be that
    // the wrong error message will show up.
    if (m.matches()) {
      if (CalciteMetadataHandler.anyTableContainsColumn(stmtTableCache, m.group(1))) {
        throw new UnsupportedFeatureException(
            "Complex column " + m.group(1) + " not supported.");
      }
    }

    m = COLUMN_NOT_FOUND.matcher(s);
    if (m.matches()) {
      if (CalciteMetadataHandler.anyTableContainsColumn(stmtTableCache, m.group(1))) {
        throw new UnsupportedFeatureException(
            "Complex column " + m.group(1) + " not supported.");
      }
    }
  }
}
